package com.yeming.flink.mudlog.function

import com.yeming.flink.mudlog.LoggingBean
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.util.Collector

/**
 * 定义一个函数类，用来保存用户上次接受数据的状态。
 */
class LoggingBeanIntervalFunction extends RichFlatMapFunction[LoggingBean, (String, LoggingBean)] {
  // 定义一个状态用于保存上一次录井数据的记录
  private var preState: ValueState[LoggingBean] = _


  override def open(parameters: Configuration): Unit = {
    preState = getRuntimeContext.getState(new ValueStateDescriptor[LoggingBean]("pre", classOf[LoggingBean]))
  }

  override def flatMap(value: LoggingBean, out: Collector[(String, LoggingBean)]): Unit = {
    var resultMsg: String = ""
    val preBean: LoggingBean = preState.value()
    if (preBean == null) {
      preState.update(value)
    } else {
      //与上一次钻取速度进行对比，如果增幅或者降幅超过200%将进行报警
      val rateInterval: Double = Math.abs((value.drillRate - preBean.drillRate) / preBean.drillRate)
      if (rateInterval > 5.0) {
        resultMsg = resultMsg + "录井钻时增幅超过600% "
      }
      // 本次井斜数据超过阈值报警
      if (value.wellDeflection > 14.95) {
        resultMsg = resultMsg + "录井井斜角度超过14.95度 "
      }
      // 与上一次井斜数据进行对比，变化超过80%报警
      val defectionInterval: Double = Math.abs((value.wellDeflection - preBean.wellDeflection) / preBean.wellDeflection)
      if (defectionInterval > 2.0) {
        resultMsg = resultMsg + "录井井斜增幅超过80% "
      }

      if (resultMsg != null && resultMsg != "") {
        value.description = resultMsg
        println(s"${value.wellName}井发现了异常数据，原因：${value.description}")
        out.collect((value.wellName, value))
        preState.update(value)
      }

    }
  }
}
